#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RocketMQ 生产消费一体化演示
同时演示消息发送和消费的完整流程
"""

import time
import json
import logging
import threading
from rocketmq.client import Producer, PushConsumer, Message, ConsumeStatus, ReceivedMessage
from config import ROCKETMQ_CONFIG, MESSAGE_CONFIG, LOGGING_CONFIG, DEMO_CONFIG, check_config, configure_rocketmq

# 配置日志
logging.basicConfig(
    level=getattr(logging, LOGGING_CONFIG["level"]), 
    format=LOGGING_CONFIG["format"]
)
logger = logging.getLogger(__name__)


class MessageHandler:
    """消息处理器"""
    
    def __init__(self):
        self.message_count = 0
        self.start_time = time.time()
        self.received_messages = []
    
    def __call__(self, msg: ReceivedMessage):
        """消息处理回调"""
        try:
            self.message_count += 1
            current_time = time.time()
            elapsed_time = current_time - self.start_time
            
            logger.info(f"=== 收到消息 #{self.message_count} (运行时间: {elapsed_time:.1f}s) ===")
            
            # 基本信息
            logger.info(f"  - Topic: {msg.topic}")
            logger.info(f"  - Tags: {msg.tags}")
            logger.info(f"  - Keys: {msg.keys}")
            logger.info(f"  - MsgId: {msg.id}")
            
            # 兼容性处理：offset 可能不存在
            if hasattr(msg, 'offset'):
                logger.info(f"  - Offset: {msg.offset}")
            
            # 消息内容
            try:
                body = msg.body.decode(MESSAGE_CONFIG["encoding"])
                logger.info(f"  - 消息内容: {body}")
                
                # 尝试解析 JSON
                try:
                    data = json.loads(body)
                    if isinstance(data, dict):
                        logger.info(f"  - 消息ID: {data.get('id', 'N/A')}")
                        logger.info(f"  - 消息类型: {data.get('data', {}).get('type', 'N/A')}")
                        logger.info(f"  - 时间戳: {data.get('timestamp', 'N/A')}")
                        
                        # 记录消息信息
                        self.received_messages.append({
                            'id': data.get('id'),
                            'msg_id': msg.id,
                            'content': data,
                            'receive_time': current_time
                        })
                except json.JSONDecodeError:
                    pass  # 不是 JSON 格式，忽略
                    
            except Exception as e:
                logger.warning(f"  - 消息内容解码失败: {e}")
                logger.info(f"  - 原始内容: {msg.body}")
            
            logger.info("=== 消息处理完成 ===")
            
            return ConsumeStatus.CONSUME_SUCCESS
            
        except Exception as e:
            logger.error(f"消息处理失败: {e}")
            return ConsumeStatus.RECONSUME_LATER


class RocketMQDemo:
    """RocketMQ 生产消费一体化演示类"""
    
    def __init__(self):
        self.producer = None
        self.consumer = None
        self.message_handler = MessageHandler()
        
        # 配置信息
        self.namesrv_addr = ROCKETMQ_CONFIG["namesrv_addr"]
        self.access_key = ROCKETMQ_CONFIG["access_key"]
        self.secret_key = ROCKETMQ_CONFIG["secret_key"]
        self.producer_group = ROCKETMQ_CONFIG["producer_group"]
        self.consumer_group = ROCKETMQ_CONFIG["consumer_group"]
        self.topic = ROCKETMQ_CONFIG["default_topic"]
        self.message_tags = ROCKETMQ_CONFIG["consumer_config"]["message_tags"]
        self.orderly = ROCKETMQ_CONFIG["consumer_config"]["orderly"]
        
        # 控制标志
        self.running = False
        self.producer_thread = None
        self.consumer_thread = None
    
    def start_producer(self):
        """启动生产者"""
        try:
            logger.info("=== 启动 RocketMQ 生产者 ===")
            logger.info(f"NameServer: {self.namesrv_addr}")
            logger.info(f"生产者组: {self.producer_group}")
            logger.info(f"主题: {self.topic}")
            
            # 创建生产者
            self.producer = Producer(self.producer_group)
            self.producer.set_name_server_address(self.namesrv_addr)
            self.producer.set_session_credentials(self.access_key, self.secret_key, "ALIYUN")
            
            # 启动生产者
            self.producer.start()
            logger.info("✓ 生产者启动成功")
            return True
            
        except Exception as e:
            logger.error(f"✗ 生产者启动失败: {e}")
            return False
    
    def start_consumer(self):
        """启动消费者"""
        try:
            logger.info("=== 启动 RocketMQ 消费者 ===")
            logger.info(f"NameServer: {self.namesrv_addr}")
            logger.info(f"消费者组: {self.consumer_group}")
            logger.info(f"主题: {self.topic}")
            logger.info(f"消息标签: {self.message_tags}")
            logger.info(f"有序消费: {self.orderly}")
            
            # 创建消费者
            self.consumer = PushConsumer(
                self.consumer_group,
                orderly=self.orderly
            )
            self.consumer.set_name_server_address(self.namesrv_addr)
            self.consumer.set_session_credentials(self.access_key, self.secret_key, "ALIYUN")
            
            # 订阅主题
            self.consumer.subscribe(self.topic, self.message_handler, self.message_tags)
            
            # 启动消费者
            self.consumer.start()
            logger.info("✓ 消费者启动成功")
            return True
            
        except Exception as e:
            logger.error(f"✗ 消费者启动失败: {e}")
            return False
    
    def send_message(self, message_data, tags=None, keys=None):
        """发送单条消息"""
        try:
            # 创建消息
            msg = Message(self.topic)
            
            # 设置消息内容
            if isinstance(message_data, dict):
                body = json.dumps(message_data, ensure_ascii=False)
            else:
                body = str(message_data)
            
            msg.set_body(body.encode(MESSAGE_CONFIG["encoding"]))
            
            # 设置消息标签
            if tags:
                msg.set_tags(tags)
            else:
                msg.set_tags(MESSAGE_CONFIG["default_tags"])
            
            # 设置消息键
            if keys:
                msg.set_keys(keys)
            
            # 发送消息
            send_result = self.producer.send_sync(msg)
            
            # 打印发送结果
            if DEMO_CONFIG["verbose"]:
                logger.info(f"✓ 消息发送成功")
                logger.info(f"  - MsgId: {send_result.msg_id}")
                logger.info(f"  - 内容: {body}")
                logger.info(f"  - 标签: {tags or MESSAGE_CONFIG['default_tags']}")
                logger.info(f"  - 键: {keys or 'N/A'}")
                
                # 兼容性处理
                if hasattr(send_result, 'offset'):
                    logger.info(f"  - 偏移量: {send_result.offset}")
                if hasattr(send_result, 'status'):
                    logger.info(f"  - 状态: {send_result.status}")
            
            return send_result
            
        except Exception as e:
            logger.error(f"✗ 消息发送失败: {e}")
            return None
    
    def producer_worker(self):
        """生产者工作线程"""
        logger.info("开始发送消息...")
        
        # 创建演示消息
        message_types = [
            {"type": "text", "content": "Hello RocketMQ!"},
            {"type": "json", "data": {"user": "demo", "action": "send"}},
            {"type": "number", "value": 42},
            {"type": "list", "items": [1, 2, 3, 4, 5]},
            {"type": "boolean", "flag": True},
        ]
        
        success_count = 0
        for i in range(DEMO_CONFIG["message_count"]):
            if not self.running:
                break
                
            try:
                msg_type = message_types[i % len(message_types)]
                msg_data = {
                    "id": i + 1,
                    "timestamp": int(time.time()),
                    "message": f"演示消息 #{i + 1}",
                    "data": msg_type,
                    "source": "python_demo"
                }
                
                result = self.send_message(
                    message_data=msg_data,
                    tags=f"demo_{i + 1}",
                    keys=f"key_{i + 1}"
                )
                
                if result:
                    success_count += 1
                
                # 消息间隔
                if i < DEMO_CONFIG["message_count"] - 1:
                    time.sleep(DEMO_CONFIG["message_interval"])
                    
            except Exception as e:
                logger.error(f"✗ 第 {i + 1} 条消息发送失败: {e}")
        
        logger.info(f"生产者完成: {success_count}/{DEMO_CONFIG['message_count']} 成功")
    
    def consumer_worker(self):
        """消费者工作线程"""
        logger.info("开始消费消息...")
        
        start_time = time.time()
        timeout = DEMO_CONFIG["consume_timeout"]
        
        while self.running and (time.time() - start_time) < timeout:
            time.sleep(1)
            
            # 每5秒打印一次状态
            elapsed = time.time() - start_time
            if int(elapsed) % 5 == 0 and elapsed > 0:
                logger.info(f"消费者运行中... 已收到 {self.message_handler.message_count} 条消息")
        
        logger.info(f"消费者完成: 共收到 {self.message_handler.message_count} 条消息")
    
    def start(self):
        """启动演示"""
        logger.info("=== 开始 RocketMQ 生产消费一体化演示 ===")
        
        try:
            # 启动生产者
            if not self.start_producer():
                return False
            
            # 启动消费者
            if not self.start_consumer():
                return False
            
            # 等待一下让消费者准备就绪
            time.sleep(2)
            
            # 设置运行标志
            self.running = True
            
            # 启动工作线程
            self.producer_thread = threading.Thread(target=self.producer_worker)
            self.consumer_thread = threading.Thread(target=self.consumer_worker)
            
            self.producer_thread.start()
            self.consumer_thread.start()
            
            # 等待线程完成
            self.producer_thread.join()
            self.consumer_thread.join()
            
            return True
            
        except Exception as e:
            logger.error(f"演示启动失败: {e}")
            return False
    
    def stop(self):
        """停止演示"""
        logger.info("正在停止演示...")
        self.running = False
    
    def shutdown(self):
        """关闭所有连接"""
        self.stop()
        
        if self.producer:
            self.producer.shutdown()
            logger.info("生产者已关闭")
        
        if self.consumer:
            self.consumer.shutdown()
            logger.info("消费者已关闭")
    
    def get_statistics(self):
        """获取统计信息"""
        return {
            "sent_messages": DEMO_CONFIG["message_count"],
            "received_messages": self.message_handler.message_count,
            "success_rate": f"{self.message_handler.message_count / DEMO_CONFIG['message_count'] * 100:.1f}%" if DEMO_CONFIG["message_count"] > 0 else "0%",
            "received_message_details": self.message_handler.received_messages
        }


def check_configuration():
    """检查配置"""
    missing_fields = check_config()
    if missing_fields:
        print(f"❌ 配置不完整，缺少: {', '.join(missing_fields)}")
        print("请先配置 RocketMQ 连接信息")
        
        if configure_rocketmq():
            print("✓ 配置完成，可以继续运行演示")
            return True
        else:
            print("❌ 配置失败，请重新运行程序")
            return False
    return True


def main():
    """主函数"""
    logger.info("开始 RocketMQ 生产消费一体化演示...")
    
    # 检查配置
    if not check_configuration():
        return
    
    # 创建演示实例
    demo = RocketMQDemo()
    
    try:
        # 启动演示
        if not demo.start():
            logger.error("演示启动失败")
            return
        
        # 等待演示完成
        logger.info("演示运行中，按 Ctrl+C 停止...")
        
        # 演示完成后等待一段时间让消费者处理完所有消息
        time.sleep(5)
        
        # 获取统计信息
        stats = demo.get_statistics()
        
        # 总结
        logger.info("=== 演示总结 ===")
        logger.info(f"发送消息数: {stats['sent_messages']}")
        logger.info(f"接收消息数: {stats['received_messages']}")
        logger.info(f"成功率: {stats['success_rate']}")
        
        if stats['received_messages'] > 0:
            logger.info("✓ 一体化演示成功完成")
            
            # 显示接收到的消息详情
            if DEMO_CONFIG["verbose"] and stats['received_message_details']:
                logger.info("=== 接收到的消息详情 ===")
                for msg in stats['received_message_details'][:5]:  # 只显示前5条
                    logger.info(f"  - 消息ID: {msg['id']}, MsgId: {msg['msg_id']}")
        else:
            logger.warning("⚠ 未收到任何消息，请检查配置和网络连接")
            
    except KeyboardInterrupt:
        logger.info("用户中断演示")
    except Exception as e:
        logger.error(f"演示过程中发生错误: {e}")
    finally:
        demo.shutdown()


if __name__ == "__main__":
    main() 